package com.atguigu.bigdata.spark.core.rdd.persist

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * create by undeRdoG on  2021-06-14  14:07
  * 凡心所向，素履以往，生如逆旅，一苇以航。
  */
object Spark05_RDD_Persist {


  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(sparkConf)

    /**
      * 一般 checkpoint 路径都是在 分布式文件存储系统中，如 HDFS
      **/
    sc.setCheckpointDir("cp")

    val list = List("Hello Scala", "Hello Spark")

    val rdd: RDD[String] = sc.makeRDD(list)

    val flatRDD = rdd.flatMap(_.split(" "))

    val mapRDD = flatRDD.map(x => {
      println("************************************************************************************")
      (x, 1)
    })


    /**
      * cache:
      * 是将 数据零时存储在内存中进行数据重用
      * 会在血缘关系中添加新的依赖，一旦出现问题，可以从头读取数据
      * persist：
      * 将临时数据存储在磁盘文件中进行数据重用，涉及到磁盘IO，性能较低，但比较安全
      * checkpoint：
      * 将数据长久的保存在磁盘中服用，涉及到磁盘IO，性能较低，但是数据安全，一般情况下，会独立的执行一个作业
      * 为了提高执行效率。所以一般情况下，需要和 cache联合使用  （这样就直接将cache中的数据写到磁盘，不会再单启一个作业）
      * 执行过程中，会切断原本的血缘关系，建立新的血缘
      * 因为将数据落在了磁盘上，相当于 成为了一个新的数据源
      **/

    mapRDD.cache()

    mapRDD.checkpoint()


    val res: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)

    res.collect().foreach(println)


    val res2: RDD[(String, Iterable[Int])] = mapRDD.groupByKey()

    res2.collect().foreach(println)

  }
}
